/*
 * Copyright (c) 2005-2007 Terracotta, Inc. All rights reserved.
 */
package org.terracotta.workmanager.pipemanager;

import java.util.Set;

import org.terracotta.workmanager.MutableWorkItem;
import org.terracotta.workmanager.queue.DefaultQueueListener;
import org.terracotta.workmanager.queue.Queue;
import org.terracotta.workmanager.queue.QueueListener;

public class DefaultPipe<T extends MutableWorkItem, ID> implements Pipe<T, ID> {

  private final Queue<T> pendingWorkQueue;
  private final Queue<T> completedWorkQueue;
  private final Queue<String> commandQueue;
  private transient QueueListener<T, ID> queueListener;
  private ID routingID;
  private String status;
  
  public static class Factory<T extends MutableWorkItem, ID> implements Pipe.Factory<T, ID> {
    
    private final Queue.Factory<T> m_queueFactory;
    private final Queue.Factory<String> m_commandQueueFactory;

    public Factory(final Queue.Factory<T> queueFactory, final Queue.Factory<String> commandQueueFactory) {
      m_queueFactory = queueFactory;
      m_commandQueueFactory = commandQueueFactory;
    }

    public Pipe<T, ID> create(ID routingID) {
      return new DefaultPipe<T, ID>(
          routingID, m_queueFactory.create(), m_queueFactory.create(), m_commandQueueFactory.create());
    }
  }

  /**
   * Private constructor - use DefaultPipe.Factory.
   *
   * @param pendingWorkQueue
   * @param completedWorkQueue
   * @param commandQueue
   */
  private DefaultPipe(final ID routingID, final Queue<T> pendingWorkQueue,
      final Queue<T> completedWorkQueue, final Queue<String> commandQueue) {
    this.routingID = routingID;
    this.pendingWorkQueue = pendingWorkQueue;
    this.completedWorkQueue = completedWorkQueue;
    this.commandQueue = commandQueue;
    this.status = "CREATED";
  }

  public Queue<T> getPendingWorkQueue() {
    return pendingWorkQueue;
  }

  public Queue<T> getCompletedWorkQueue() {
    return completedWorkQueue;
  }

  public Queue<String> getCommandQueue() {
    return commandQueue;
  }
  
  public String getStatus() {
    return status;
  }
  
  public void setStatus(String status) {
    synchronized(status) {
      this.status = status;
    }
  }

  public ID getRoutingID() {
    return routingID;
  }

  public QueueListener<T, ID> getQueueListener() {
    return queueListener;
  }

  public void startQueueListener(final Set<T> allCompletedWork) {
    if (queueListener != null) return;
    queueListener =
        new DefaultQueueListener<T, ID>(routingID, completedWorkQueue, allCompletedWork);
    queueListener.start();
  }

  public void clear() {

    if (queueListener != null) {
      queueListener.stop();
    }

    pendingWorkQueue.clear();
    completedWorkQueue.clear();
    commandQueue.clear();

    try {
      Thread.sleep(100);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }
}
